package com.we.risk.plist;

import org.apache.flink.api.common.cache.DistributedCache;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.we.flink.utils.WeKafkaPropertyReader;
import com.we.risk.plist.utils.PlistTools;
import com.we.utils.CommonTimeTools;
import com.we.utils.ParamCheck;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.*;

import static com.we.risk.plist.utils.PlistTools.*;

public class AdmPlistLabelTest {
    public static final String RELEASEPROP = "risk/plist/adm_plist_kfk_self_test.properties";
    public static final String PROPSETS = "risk/plist/plist20230406.properties";
    public static Logger LOG = LoggerFactory.getLogger(AdmPlistLabelTest.class);

    public static final String APPTAGDATA = "hdfs:///data/dim/app_data_tag/apk_tag_data";
    public static final String APPTAGFILENAME = "app_apk_tag";
    public static final String HFQAPPNAME = "好分期";

    public static final String SPLITWORD = "#";
    public static final String DATEFORMAT = "yyyy-MM-dd";
    public static final String MSGTYPE = "HFQ_UPLOAD_PLIST_RECORD";
    public static final String INPUTKEYBY = "userKey";
    public static final String OUTPUTKEY = "user_key";
    public static final String HFQINSTIME = "hfq_install_time";
    public static final String HTTPTIMESTAMP = "httpTimestamp";
    public static final String CAPTIME = "capture_time";
    public static final String LABELOUTPUTIME = "label_output_time";
    public static final String JOBID = "jobID";
    public static final String APPLIST = "appList";
    public static final String LASTOPENTIME = "lastOpenTime";
    public static final String LASTUPDATETIME = "lastUpdateTime";
    public static final String LASTISTIME = "lastInstallTime";
    public static final String APPNAME = "appName";
    public static final String APKNAME = "apkName";
    public static final String PACKAGENAME = "packageName";
    public static final String AFTHFQ = "is_aft_hfq";
    public static final String BFHFQ = "is_bf_hfq";
    public static final long INVAILDTIME = 315417600000L;

    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();

            Properties consumProp = new Properties();
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            /** distributed catch */
            env.registerCachedFile(APPTAGDATA, APPTAGFILENAME);

            /** set offset to latest */
            KafkaSource<String> kfkSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(srcKafkaBootStrapServer)
                            .setTopics(srcKafkaTopic)
                            .setStartingOffsets(OffsetsInitializer.earliest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            SingleOutputStreamOperator<String> input =
                    env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "Plist")
                            .uid("kfk-Plist-source");

            SingleOutputStreamOperator<String> out =
                    input.flatMap(new PlistProcessRichFlatMapFunc())
                            .keyBy(
                                    new KeySelector<String, String>() {
                                        @Override
                                        public String getKey(String value) throws Exception {
                                            JSONObject inputJson = JSONObject.parseObject(value);
                                            return inputJson.getString(OUTPUTKEY);
                                        }
                                    })
                            .flatMap(new PlistRichFlatMapFunc())
                            .keyBy(
                                    new KeySelector<String, String>() {
                                        @Override
                                        public String getKey(String value) throws Exception {
                                            JSONObject inputJson = JSONObject.parseObject(value);
                                            return inputJson.getString(OUTPUTKEY);
                                        }
                                    })
                            .flatMap(
                                    new RichFlatMapFunction<String, String>() {
                                        ValueState<JSONObject> plistVState;

                                        @Override
                                        public void open(Configuration parameters)
                                                throws Exception {
                                            plistVState =
                                                    getRuntimeContext()
                                                            .getState(
                                                                    new ValueStateDescriptor<
                                                                            JSONObject>(
                                                                            "PlistVState",
                                                                            JSONObject.class));
                                        }

                                        @Override
                                        public void flatMap(String value, Collector<String> out)
                                                throws Exception {
                                            try {
                                                JSONObject inputJson =
                                                        JSONObject.parseObject(value);
                                                JSONObject tmpState = plistVState.value();
                                                JSONObject curState =
                                                        tmpState == null
                                                                ? initPropJson(PROPSETS)
                                                                : tmpState;
                                                String outKey = inputJson.getString(OUTPUTKEY);
                                                Long input_cap_time = inputJson.getLong(CAPTIME);
                                                Long state_cap_time = curState.getLong(CAPTIME);
                                                if (input_cap_time > state_cap_time) {
                                                    Iterator<String> iterator =
                                                            curState.keySet().iterator();
                                                    while (iterator.hasNext()) {
                                                        String key = iterator.next();
                                                        if (!key.equals(OUTPUTKEY)
                                                                && !key.equals(CAPTIME)) {
                                                            Object labelV = inputJson.get(key);
                                                            Object state_v = curState.get(key);
                                                            if (state_v != null && labelV != null) {
                                                                if (!state_v.equals(labelV)) {
                                                                    JSONObject outJson =
                                                                            new JSONObject();
                                                                    outJson.put(OUTPUTKEY, outKey);
                                                                    outJson.put(
                                                                            CAPTIME,
                                                                            input_cap_time);
                                                                    outJson.put("label_name", key);
                                                                    outJson.put(
                                                                            "label_value", labelV);
                                                                    out.collect(outJson.toString());
                                                                    curState.put(key, labelV);
                                                                }
                                                            } else {
                                                                if (state_v != labelV) {
                                                                    JSONObject outJson =
                                                                            new JSONObject();
                                                                    outJson.put(OUTPUTKEY, outKey);
                                                                    outJson.put(
                                                                            CAPTIME,
                                                                            input_cap_time);
                                                                    outJson.put("label_name", key);
                                                                    outJson.put(
                                                                            "label_value", labelV);
                                                                    out.collect(
                                                                            outJson.toString(
                                                                                    SerializerFeature
                                                                                            .WriteMapNullValue));
                                                                    curState.put(key, labelV);
                                                                }
                                                            }
                                                        }
                                                    }
                                                    curState.put(CAPTIME, input_cap_time);
                                                    plistVState.update(curState);
                                                }
                                            } catch (Exception e) {
                                                LOG.error("FlatMap Exception: " + e.toString());
                                                LOG.error("FlatMap error input: " + value);
                                            }
                                        }
                                    });

            /** sink to Kafka */
            String sinkkfkTopic = paramReader.getKfkTopic();
            Properties sinkProp = new Properties();
            sinkProp.setProperty("bootstrap.servers", paramReader.getKfkBootStrapServer());
            //            sinkProp.setProperty("max.request.size", String.valueOf(KFKMSGMAXSIZE));
            int sinkkfkPartitions = paramReader.getKfkPartitions();

            FlinkKafkaProducer<String> kafkaProducer =
                    new FlinkKafkaProducer<String>(
                            sinkkfkTopic,
                            new WeKafkaKeyedSerializationSchema(),
                            sinkProp,
                            Optional.of(new WeKafkaCustomPartitioner()));

            out.addSink(kafkaProducer).setParallelism(sinkkfkPartitions);

            env.execute(AdmPlistLabelTest.class.toString());
        } catch (Exception e) {
            System.out.println("Exception: " + e.getMessage());
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static class PlistProcessRichFlatMapFunc extends RichFlatMapFunction<String, String> {

        @Override
        public void flatMap(String input, Collector<String> out) throws Exception {
            try {
                if (input.contains(MSGTYPE)) {
                    JSONObject inputJson = JSONObject.parseObject(input);
                    if (inputJson != null && inputJson.getString("type").equals(MSGTYPE)) {
                        JSONObject dataJson = inputJson.getJSONObject("dataJson");
                        ParamCheck.isNull(dataJson, "dataJson");
                        JSONObject job = dataJson.getJSONObject("job");
                        String jobID = job.getString(JOBID);
                        ParamCheck.isNull(jobID, "jobID");

                        String userKey = dataJson.getString(INPUTKEYBY);
                        ParamCheck.isNull(userKey, INPUTKEYBY);

                        JSONObject data = dataJson.getJSONObject("data");
                        JSONObject baseData = data.getJSONObject("baseData");
                        ParamCheck.isNull(baseData, "baseData");
                        Long captureTime = baseData.getLong(HTTPTIMESTAMP);
                        ParamCheck.isNull(captureTime, "captureTime");

                        Integer isAuthorized = baseData.getInteger("isAuthorized");
                        if (isAuthorized != null && isAuthorized != 3) {
                            JSONObject outJson = appListFilter(data);
                            outJson.put(OUTPUTKEY, userKey);
                            outJson.put(CAPTIME, captureTime);
                            outJson.put(JOBID, jobID);
                            out.collect(outJson.toString());
                        }
                    }
                }
            } catch (Exception e) {
                LOG.error(e.toString());
                LOG.error("flatMap1 error input: " + input);
            }
        }

        private JSONObject appListFilter(JSONObject input) {
            JSONObject res = new JSONObject();
            JSONArray jsonArray = input.getJSONArray(APPLIST);
            int size = jsonArray.size();
            JSONArray resArray = new JSONArray();
            for (int i = 0; i < size; i++) {
                JSONObject callDetail = (JSONObject) jsonArray.get(i);
                String installTime = callDetail.getString(LASTOPENTIME);
                if (installTime != null && !installTime.equals("")) {
                    String apkName = callDetail.getString(PACKAGENAME);
                    callDetail.remove(PACKAGENAME);
                    callDetail.put(APKNAME, apkName);

                    String appName = callDetail.getString(APPNAME);
                    if (appName != null && appName.equals(HFQAPPNAME)) {
                        //                        hfqInstallTime = callDetail.getLong(installTime);
                        res.put(HFQINSTIME, installTime);
                    }
                    callDetail.remove(LASTOPENTIME);
                    callDetail.put(LASTISTIME, installTime);
                    resArray.add(callDetail);
                }
            }
            res.put(APPLIST, resArray);
            return res;
        }
    }

    private static class PlistRichFlatMapFunc extends RichFlatMapFunction<String, String> {
        SimpleDateFormat dateFormat;
        HashMap<String, ArrayList<String>> appTagMap;
        HashMap<String, ArrayList<String>> apkTagMap;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            appTagMap = new HashMap<>();
            apkTagMap = new HashMap<>();
            dateFormat = new SimpleDateFormat(DATEFORMAT);
            initTagMap();
        }

        @Override
        public void flatMap(String input, Collector<String> out) throws Exception {
            try {
                HashMap<String, AppCategoryBasicInfo> cateMap = initMap();
                AppCategoryBasicInfo totInfo = cateMap.get(TAGTOT);

                JSONObject inputJson = JSONObject.parseObject(input);
                Long cap_time = inputJson.getLong(CAPTIME);
                String user_key = inputJson.getString(OUTPUTKEY);
                String job_id = inputJson.getString(JOBID);
                Long hfqInstTime = inputJson.getLong(HFQINSTIME);
                JSONArray app_list = inputJson.getJSONArray(APPLIST);
                int size = app_list.size();
                for (int i = 0; i < size; i++) {
                    JSONObject appDetail = (JSONObject) app_list.get(i);
                    //                    LOG.warn("appDetail: " + appDetail.toString());
                    String apk_name = appDetail.getString(APKNAME);
                    String app_name = appDetail.getString(APPNAME);
                    String app_apk =
                            (app_name == null ? "" : app_name)
                                    + "|"
                                    + (apk_name == null ? "" : apk_name);
                    byte loanTag = PlistTools.getLoanTag(app_name, apk_name);
                    byte bankTag = PlistTools.getBankTag(app_name);
                    byte betTag = PlistTools.getBetTag(app_name, apk_name);
                    Long install_time = appDetail.getLong(LASTISTIME);
                    String inst_hfq_tag = null;
                    if (hfqInstTime != null) {
                        inst_hfq_tag = (install_time > hfqInstTime ? AFTHFQ : BFHFQ);
                    } else {
                        inst_hfq_tag = "NONE";
                    }
                    //                    String inst_hfq_tag = (install_time > hfqInstTime ? AFTHFQ
                    // : BFHFQ);
                    String inst_h_part = getHourPart(install_time);
                    Long update_time = appDetail.getLong(LASTUPDATETIME);
                    String updt_h_part = getHourPart(update_time);

                    ArrayList<String> appTags = getAppTag(app_name, apk_name, appTagMap, apkTagMap);
                    /** INVAILDTIME 返回NULL */
                    Integer inst_day = dateBetween(cap_time, install_time, dateFormat);
                    Integer inst_mth = monthBetween(cap_time, install_time);
                    Integer updt_day = dateBetween(cap_time, update_time, dateFormat);
                    Integer updt_mth = monthBetween(cap_time, update_time);

                    Integer inst_day_cmp_hfq = dateBetween(install_time, hfqInstTime, dateFormat);

                    AppApkObject appApkObj =
                            new AppApkObject(
                                    app_apk,
                                    inst_h_part,
                                    updt_h_part,
                                    inst_day,
                                    inst_day_cmp_hfq,
                                    updt_day,
                                    inst_mth,
                                    updt_mth,
                                    appTags,
                                    inst_hfq_tag);

                    //                    LOG.warn("Debug appApkObj: " + appApkObj.toString());

                    InsertApkToCategory(totInfo, appApkObj);

                    if (loanTag == 1) {
                        InsertApkToCategory(cateMap.get(TAGLOAN), appApkObj);
                    }
                    if (bankTag == 1) {
                        InsertApkToCategory(cateMap.get(TAGBANK), appApkObj);
                    }
                    if (betTag == 1) {
                        InsertApkToCategory(cateMap.get(TAGBET), appApkObj);
                    }
                    for (String appTag : appTags) {
                        InsertApkToCategory(cateMap.get(appTag), appApkObj);
                    }
                }

                JSONObject resJson = new JSONObject();
                resJson.put(OUTPUTKEY, user_key);
                resJson.put(CAPTIME, cap_time);
                resJson.put("job_id", job_id);
                totInfo.calcuTotalLabel();
                Set<String> keySet = cateMap.keySet();
                for (String k : keySet) {
                    AppCategoryBasicInfo basicInfo = cateMap.get(k);
                    String cate = basicInfo.getCate();
                    HashMap<String, Object> singleMap = basicInfo.getSingleMap();
                    if (!k.equals(TAGTOT)) {
                        if (basicInfo != null) {
                            basicInfo.calcuCateLabel(totInfo.getBasicMap());
                        }
                    }
                    Set<String> keySets = singleMap.keySet();
                    for (String label : keySets) {
                        //                        String suffix = label.equals("") ? "cnt" : label;
                        String suffix = "";
                        if (label.equals("")) {
                            suffix = "cnt";
                        } else if (label.equals("bf_hfq") || label.equals("aft_hfq")) {
                            suffix = label + "_cnt";
                        } else {
                            suffix = label;
                        }
                        resJson.put(cate + "_" + suffix, singleMap.get(label));
                    }
                }
                long output_time = System.currentTimeMillis();
                resJson.put(LABELOUTPUTIME, output_time);
                out.collect(resJson.toString());
            } catch (Exception e) {
                LOG.error(e.toString());
                LOG.error("PlistRichFlatMapFunc error input: " + input);
            }
        }

        private void initTagMap() throws IOException {
            DistributedCache distributedCache = getRuntimeContext().getDistributedCache();
            File file = distributedCache.getFile(APPTAGFILENAME);
            List<String> lines = FileUtils.readLines(file);
            for (String line : lines) {
                String[] splits = line.split(",");
                if (splits.length == 3) {
                    String appName = splits[0].toLowerCase();
                    String apkName = splits[1].toLowerCase();
                    String tag = splits[2];
                    ArrayList<String> appTagArray = appTagMap.get(appName);
                    ArrayList<String> apkTagArray = apkTagMap.get(apkName);
                    if (appTagArray == null) {
                        appTagArray = new ArrayList<>();
                    }
                    if (apkTagArray == null) {
                        apkTagArray = new ArrayList<>();
                    }
                    appTagArray.add(tag);
                    apkTagArray.add(tag);

                    appTagMap.put(appName, appTagArray);
                    apkTagMap.put(apkName, apkTagArray);
                }
            }
        }

        //        private void processLabel(AppCategoryBasicInfo info, JSONObject jsonObject, String
        // tag){
        //            HashMap<String, HashSet<String>> basicMap = info.getBasicMap();
        //            Set<String> keySets = basicMap.keySet();
        //
        //            for (String key : keySets) {
        //                int cnt = basicMap.get(key).size();
        //                if(tag.equals("")) {
        //                    if(key.equals("")) {
        //                        jsonObject.put("tot_cnt", cnt);
        //                    } else {
        //                        key = "tot_"+key;
        //                        jsonObject.put(key, cnt);
        //                    }
        //                } else {
        //                    key = tag+"_"+key;
        //                    jsonObject.put(key, cnt);
        //                }
        //
        //            }
        //        }

        private HashMap<String, AppCategoryBasicInfo> initMap() {
            HashMap<String, AppCategoryBasicInfo> map = new HashMap<>();
            map.put(TAGTOT, new AppCategoryBasicInfo(TAGTOT));
            map.put(TAGBANK, new AppCategoryBasicInfo(TAGBANK));
            map.put(TAGLOAN, new AppCategoryBasicInfo(TAGLOAN));
            map.put(TAGBET, new AppCategoryBasicInfo(TAGBET));
            map.put(TAGJJSH, new AppCategoryBasicInfo(TAGJJSH));
            map.put(TAGJRLC, new AppCategoryBasicInfo(TAGJRLC));
            map.put(TAGLTSJ, new AppCategoryBasicInfo(TAGLTSJ));
            map.put(TAGLYJT, new AppCategoryBasicInfo(TAGLYJT));
            map.put(TAGNONE, new AppCategoryBasicInfo(TAGNONE));
            map.put(TAGSCTY, new AppCategoryBasicInfo(TAGSCTY));
            map.put(TAGSJYX, new AppCategoryBasicInfo(TAGSJYX));
            map.put(TAGSSGW, new AppCategoryBasicInfo(TAGSSGW));
            map.put(TAGSYGJ, new AppCategoryBasicInfo(TAGSYGJ));
            map.put(TAGSYSX, new AppCategoryBasicInfo(TAGSYSX));
            map.put(TAGTSYD, new AppCategoryBasicInfo(TAGTSYD));
            map.put(TAGXLBG, new AppCategoryBasicInfo(TAGXLBG));
            map.put(TAGXWZX, new AppCategoryBasicInfo(TAGXWZX));
            map.put(TAGXXJY, new AppCategoryBasicInfo(TAGXXJY));
            map.put(TAGXXYL, new AppCategoryBasicInfo(TAGXXYL));
            map.put(TAGYLJK, new AppCategoryBasicInfo(TAGYLJK));
            map.put(TAGYYST, new AppCategoryBasicInfo(TAGYYST));
            map.put(TAGOTHER, new AppCategoryBasicInfo(TAGOTHER));
            return map;
        }
    }

    private static void InsertApkToCategory(AppCategoryBasicInfo cateInfo, AppApkObject appApkObj) {
        String cate = cateInfo.getCate();
        cateInfo.addKeySet(AppCategoryBasicInfo.ALLSET, appApkObj.getApp_apk());
        // list
        cateInfo.addInstDay(appApkObj.getInst_day());
        cateInfo.addInstMth(appApkObj.getInst_mth());
        cateInfo.addUpdtDay(appApkObj.getUpdt_day());
        cateInfo.addUpdtMth(appApkObj.getUpdt_mth());
        // set
        cateInfo.addHourPartSet(appApkObj.getApp_apk(), appApkObj.getInst_h_part(), "inst");
        cateInfo.addHourPartSet(appApkObj.getApp_apk(), appApkObj.getUpdt_h_part(), "updt");
        cateInfo.addDayPartSet(appApkObj.getApp_apk(), appApkObj.getInst_day(), "inst");
        cateInfo.addDayPartSet(appApkObj.getApp_apk(), appApkObj.getUpdt_day(), "updt");
        cateInfo.addMonPartSet(
                appApkObj.getApp_apk(), appApkObj.getInst_mth(), "inst", cateInfo.getCate());
        cateInfo.addMonPartSet(
                appApkObj.getApp_apk(), appApkObj.getUpdt_mth(), "updt", cateInfo.getCate());

        // 银行和贷款类没有month range
        if (cate.equals(TAGBANK) || cate.equals(TAGLOAN)) {
            if (appApkObj.getInst_hfq_tag() != null) {
                String tmpTag = appApkObj.getInst_hfq_tag();
                if (tmpTag.equals(BFHFQ)) {
                    tmpTag = "bf_hfq";
                } else if (tmpTag.equals(AFTHFQ)) {
                    tmpTag = "aft_hfq";
                }
                cateInfo.addKeySet(tmpTag, appApkObj.getApp_apk());
                /** add patch */
                cateInfo.addInstDayCmpHfqSet(tmpTag, appApkObj);

                cateInfo.addDayPartSet(
                        appApkObj.getApp_apk(),
                        appApkObj.getInst_day(),
                        appApkObj.getInst_hfq_tag() + "_inst");
                cateInfo.addDayPartSet(
                        appApkObj.getApp_apk(),
                        appApkObj.getUpdt_day(),
                        appApkObj.getInst_hfq_tag() + "_updt");
                cateInfo.addMonPartSet(
                        appApkObj.getApp_apk(),
                        appApkObj.getInst_mth(),
                        appApkObj.getInst_hfq_tag() + "_inst",
                        cateInfo.getCate());
                cateInfo.addMonPartSet(
                        appApkObj.getApp_apk(),
                        appApkObj.getUpdt_mth(),
                        appApkObj.getInst_hfq_tag() + "_updt",
                        cateInfo.getCate());
            }
        }
    }

    private static Integer dateBetween(long base_time, Long com_time, SimpleDateFormat sdf)
            throws Exception {
        if (com_time == null || com_time <= INVAILDTIME) {
            //            return CommonTimeTools.dateDiff(captime, INVAILDTIME, sdf);
            return null;
        }

        return CommonTimeTools.dateDiff(base_time, com_time, sdf);
    }

    private static Integer monthBetween(Long base_time, Long com_time) {
        if (com_time == null || com_time <= INVAILDTIME) {
            //            return CommonTimeTools.monthDiff(captime, INVAILDTIME);
            return null;
        }
        return CommonTimeTools.monthDiff(base_time, com_time);
    }

    private static JSONObject initPropJson(String propFile) throws IOException {
        InputStream resourceAsStream =
                AdmPlistLabelTest.class.getClassLoader().getResourceAsStream(propFile);
        ParameterTool reader = ParameterTool.fromPropertiesFile(resourceAsStream);
        JSONObject res = new JSONObject();
        Iterator<Object> keysIterator = reader.getProperties().keySet().iterator();
        while (keysIterator.hasNext()) {
            String key = (String) keysIterator.next();
            if (key.equals(CAPTIME)) {
                res.put(key, -1);
            } else {
                res.put(key, "initV");
            }
        }
        return res;
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String keyby = jsonObject.getString(OUTPUTKEY);
            return keyby.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.info(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }
}
